package com.xx.rpc.core.remoting.net.params;

import com.xx.rpc.core.remoting.invoker.XxRpcInvokerFactory;
import com.xx.rpc.core.remoting.invoker.call.XxRpcInvokeCallback;
import com.xx.rpc.core.util.XxRpcException;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * rpc异步响应
 *
 * @author hanxiyue
 */
public class XxRpcFutureResponse implements Future<XxRpcResponse> {

    //消费端工厂
    private XxRpcInvokerFactory invokerFactory;

    // rpc请求
    private XxRpcRequest request;

    //rpc响应
    private XxRpcResponse response;

    private boolean done = false;
    private Object lock = new Object();

    //消费端回调类
    private XxRpcInvokeCallback invokeCallback;
    // ---------------------- 响应池 ----------------------

    public void setInvokerFuture() {
        this.invokerFactory.setInvokerFuture(request.getRequestId(), this);
    }

    public void removeInvokerFuture() {
        this.invokerFactory.removeInvokerFuture(request.getRequestId());
    }

    // ---------------------- get ----------------------
    public XxRpcRequest getRequest() {
        return request;
    }
    public XxRpcInvokeCallback getInvokeCallback() {
        return invokeCallback;
    }
    // ---------------------- 设置消费端 ----------------------

    public void setResponse(XxRpcResponse response) {
        this.response = response;
        synchronized (lock) {
            done = true;
            lock.notifyAll();
        }
    }

    // ---------------------- 设置消费者 ----------------------
    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return false;
    }

    @Override
    public boolean isCancelled() {
        return false;
    }

    @Override
    public boolean isDone() {
        return false;
    }

    /**
     * 获取响应参数
     *
     * @return 响应参数
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @Override
    public XxRpcResponse get() throws InterruptedException, ExecutionException {
        try {
            return get(-1, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            throw new XxRpcException(e);
        }
    }

    /**
     * 获取响应参数
     *
     * @param timeout 超时时间
     * @param unit    单位
     * @return 响应参数
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws TimeoutException
     */
    @Override
    public XxRpcResponse get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        if (!done) {//判断是否设置完
            synchronized (lock) {
                try {
                    if (timeout < 0) {
                        lock.wait();
                    } else {
                        long timeoutMillis = (TimeUnit.MILLISECONDS == unit) ? timeout : TimeUnit.MILLISECONDS.convert(timeout, unit);
                        lock.wait(timeoutMillis);
                    }
                } catch (InterruptedException e) {
                    throw e;
                }
            }
        }

        if (!done) {
            throw new XxRpcException("xx-rpc, 请求超时:" + System.currentTimeMillis() + ", request:" + request.toString());
        }
        return response;
    }
}
